package com.originskyseed.platform.api.service.impl;

import java.math.BigDecimal;
import java.net.InetAddress;
import java.util.*;
import java.util.stream.Collectors;

import com.originskyseed.platform.common.api.util.DistributedLockUtil;
import com.originskyseed.platform.core.redis.util.RedisUtil;
import com.originskyseed.platform.user.model.*;
import com.originskyseed.platform.user.service.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.aspectj.apache.bcel.util.Play;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.originskyseed.platform.core.rocketmq.util.RocketMQPropertiesUtil;
import com.originskyseed.platform.core.tenant.util.TenantCodeUtil;
import com.originskyseed.platform.core.util.bean.PrimaryKeyGeneratorUtil;

import static java.lang.Math.random;

public class MqMsgLoadListener implements ApplicationListener<ContextRefreshedEvent>, DisposableBean {
	private static final Logger logger = LoggerFactory.getLogger(MqMsgLoadListener.class);

	@DubboReference
	private IAnchorService anchorService;

	@DubboReference
	private IAnchorRelateAnnexService anchorRelateAnnexService;

	@DubboReference
	private IPlayerService playerService;

	@DubboReference
	private IAnchorRelateLiveStateService anchorRelateLiveStateService;

	@DubboReference
	private IAnchorRelateLiveService anchorRelateLiveService;

	@DubboReference
	private IPlayerRelateConsumeService playerRelateConsumeService;

	@DubboReference
	private IPlayerRelateLoginService playerRelateLoginService;

	@DubboReference
	private IPlayerRelateRechargeService playerRelateRechargeService;

	@DubboReference
	private IPlayerRelateViewLiveStateService playerRelateViewLiveStateService;

	@DubboReference
	private IPlayerRelateViewLiveService playerRelateViewLiveService;

	@DubboReference
	private IAnchorRelateInteractService anchorRelateInteractService;

	@DubboReference
	private IAnchorRelateByoFansService anchorRelateByoFansService;

	@DubboReference
	private IPlayerRelateChannelLogService playerRelateChannelLogService;

	@DubboReference
	private IPlayerRelateRechargeMonthService playerRelateRechargeMonthService;

	@DubboReference
	private IPlayerRelateLastLoginService playerRelateLastLoginService;

	@DubboReference
	private IPlayerRelateAnchorRechargeService playerRelateAnchorRechargeService;

	@DubboReference
	private IUserCumulativeChargeService userCumulativeChargeService;

	@DubboReference
	private IParameterConfigureService parameterConfigureService;

	@DubboReference
	private IThreeLoginUserService threeLoginUserService;

	@DubboReference
	private ISevenLoginUserService sevenLoginUserService;

	private DefaultMQPushConsumer consumer;
	/*
	 * 配置文件实例
	 */
	private static Properties props = null;

	@Override
	public void onApplicationEvent(ContextRefreshedEvent event) {
		System.out.println("=Mq=onApplicationEvent==");
		// 如果是第二次加载则不处理了
		if (event.getApplicationContext().getParent() != null) {
			return;
		}
		createConsumer();
	}

	/**
	 * 创建消费者
	 */
	private void createConsumer() {
		// 初始化并启动
		try {
			props = RocketMQPropertiesUtil.getInstance().getProperties();
			consumer = new DefaultMQPushConsumer("PushConsumer");
			// 服务发现和路由地址
			String namesrvAddr = props.getProperty("rocketmq.namesrvAddr");
			String retryTimesStr = props.getProperty("rocketmq.producer.retryTimes");

			if (namesrvAddr != null && namesrvAddr.trim().length() != 0) {
				consumer.setNamesrvAddr(namesrvAddr);
			}
			consumer.subscribe("synchroData", "push");
			// 程序第一次启动从消息队列头取数据
			consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
			consumer.registerMessageListener(new MessageListenerConcurrently() {
				public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext Context) {
					MessageExt msg = list.get(0);
					String topic = msg.getTopic();
					byte[] body = msg.getBody();
					String keys = msg.getKeys();
					String content = new String(body);

					int retryTimes = 3;
					if (retryTimesStr != null && StringUtils.isNotEmpty(retryTimesStr)) {
						retryTimes = Integer.parseInt(retryTimesStr);
					}
					if (msg.getReconsumeTimes() == retryTimes) { // 消息重发了三次
						// 持久化 消息记录表
						return ConsumeConcurrentlyStatus.RECONSUME_LATER; // 签收
					}
					JSONObject jsonObject = JSONObject.parseObject(content);
					String tenantCode = jsonObject.getString("tenantCode");
					content = jsonObject.getString("list");

					ConsumeConcurrentlyStatus consumeConcurrentlyStatus = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					switch (keys) {
					case "playerInsert":
						consumeConcurrentlyStatus = playerInsert(tenantCode,content);
						break;
					case "playerRelateConsumeInsert":
						consumeConcurrentlyStatus = playerRelateConsumeInsert(tenantCode,content);
						break;
					case "playerRelateLoginInsert":
						consumeConcurrentlyStatus = playerRelateLoginInsert(tenantCode,content);
						break;
					case "playerRelateRechargeInsert":
						consumeConcurrentlyStatus = playerRelateRechargeInsert(tenantCode,content);
						break;
					case "playerRelateViewLiveInsert":
						consumeConcurrentlyStatus = playerRelateViewLiveInsert(tenantCode,content);
						break;
					case "playerRelateChannelLogInsert":
						consumeConcurrentlyStatus = playerRelateChannelLogInsert(tenantCode,content);
						break;
					case "anchorInsert":
						consumeConcurrentlyStatus = anchorInsert(tenantCode,content);
						break;
					case "anchorLiveInsert":
						consumeConcurrentlyStatus = anchorRelateLiveInsert(tenantCode,content);
						break;
					case "anchorRelateInteractInsert":
						consumeConcurrentlyStatus = anchorRelateInteractInsert(tenantCode,content);
						break;
					case "anchorRelateByoFansInsert":
						consumeConcurrentlyStatus = anchorRelateByoFansInsert(tenantCode,content);
						break;
					default:
						break;
					}
					return consumeConcurrentlyStatus;
				}
			});
			consumer.start();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 服务停止
	 */
	@Override
	public void destroy() throws Exception {
		consumer.shutdown();
	}

	/**
	 * 新增玩家
	 * 
	 * @param content
	 * @return
	 */
	private ConsumeConcurrentlyStatus playerInsert(String tenantCode,String content) {
		try {
			if(tenantCode!=null&&StringUtils.isNotBlank(tenantCode))
			{
				TenantCodeUtil.getInstance().createServerTenantCode(tenantCode);
			}

			List<Player> list = JSONArray.parseArray(content, Player.class);
			if (list != null && list.size() > 0) {
				// 新增的玩家集合
				List<Player> insertPlayer = new ArrayList<>();
				// 查询有没有重复的玩家 有则去重

				List<String> playerIds = list.stream().map(Player::getPlayerId).collect(Collectors.toList());

				//为了防止多节点时重复插入相同的记录数据
				String lockKey="playerInsert_"+String.join(",",playerIds);
				RedisUtil.getInstance().lock(lockKey,1500L);

				playerIds = this.playerService.selectPlayerIds(tenantCode,playerIds);
				for (Player player : list) {
					boolean isInsert = true;
					for (String id : playerIds) {
						if (player.getPlayerId().equals(id)) {
							isInsert = false;
						}
					}
					if (isInsert) {
						//player.setBelongBusinessId(player.getChannelSubId());
						player.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
						player.setCreateTime(new Date());
						insertPlayer.add(player);
					}
				}

				// 新增玩家 和用户
				if (insertPlayer.size() > 0) {
					// 一次性插入的数量
					int size = 500;
					// 总长度
					int userLength = insertPlayer.size();
					// 计算可以分成多少组
					int num = (userLength + size - 1) / size;
					for (int i = 0; i < num; i++) {
						// 开始位置
						int fromIndex = i * size;
						// 结束位置
						int toIndex = (i + 1) * size < userLength ? (i + 1) * size : userLength;

						List<Player> newPlayerList = insertPlayer.subList(fromIndex, toIndex);
						this.playerService.insertBatch(tenantCode,newPlayerList);
					}
				}
			}
		} catch (Exception ex) {

			logger.error(ex.getMessage());
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		} finally {
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	}


	/**
	 * 新增主播
	 *
	 * @param content
	 * @return
	 */
	private ConsumeConcurrentlyStatus anchorInsert(String tenantCode,String content) {
		try {
			if(tenantCode!=null&&StringUtils.isNotBlank(tenantCode))
			{
				TenantCodeUtil.getInstance().createServerTenantCode(tenantCode);
			}
			List<Anchor> list = JSONArray.parseArray(content, Anchor.class);
			if (list != null && list.size() > 0) {
				// 新增的主播集合
				List<Anchor> insertAnchorList = new ArrayList<>();
				List<AnchorRelateAnnex> anchorRelateAnnexList = new ArrayList<>();

				// 查询有没有重复的主播 有则去重
				List<String> anchorIds = list.stream().map(Anchor::getAnchorId).collect(Collectors.toList());

				//为了防止多节点时重复插入相同的记录数据
				String lockKey="anchorInsert_"+String.join(",",anchorIds);
				RedisUtil.getInstance().lock(lockKey,1000L);

				anchorIds = anchorService.selectAnchorIds(anchorIds);

				for (Anchor anchor : list) {
					boolean isInsert = true;
					for (String anchorId : anchorIds) {
						if (anchorId.equals(anchor.getAnchorId())) {
							isInsert = false;
						}
					}
					if (isInsert) {
						anchor.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
						anchor.setIsDel(0);
						anchor.setCreateTime(new Date());
						if (StringUtils.isNotEmpty(anchor.getAnchorMedal())) {
							AnchorRelateAnnex anchorRelateAnnex = new AnchorRelateAnnex();
							anchorRelateAnnex.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
							anchorRelateAnnex.setAnchorId(anchor.getId());
							anchorRelateAnnex.setPath(anchor.getAnchorMedal());
							anchorRelateAnnex.setType(1);
							anchorRelateAnnexList.add(anchorRelateAnnex);
						}
						insertAnchorList.add(anchor);
					}
				}

				// 一次性插入的数量
				int size = 500;
				// 新增主播和用户信息
				if (insertAnchorList.size() > 0) {
					// 总长度
					int userLength = insertAnchorList.size();
					// 计算可以分成多少组
					int num = (userLength + size - 1) / size;
					for (int i = 0; i < num; i++) {
						// 开始位置
						int fromIndex = i * size;
						// 结束位置
						int toIndex = (i + 1) * size < userLength ? (i + 1) * size : userLength;
						List<Anchor> newAnchorList = insertAnchorList.subList(fromIndex, toIndex);
						this.anchorService.insertBatchAnchor(newAnchorList);
					}
				}

				// 新增主播附件信息
				if (anchorRelateAnnexList.size() > 0) {
					// 总长度
					int length = anchorRelateAnnexList.size();
					// 计算可以分成多少组
					int num = (length + size - 1) / size;
					for (int i = 0; i < num; i++) {
						// 开始位置
						int fromIndex = i * size;
						// 结束位置
						int toIndex = (i + 1) * size < length ? (i + 1) * size : length;
						List<AnchorRelateAnnex> newAnchorRelateAnnex = anchorRelateAnnexList.subList(fromIndex, toIndex);
						anchorRelateAnnexService.insertBatch(newAnchorRelateAnnex);
					}
				}
			}
		} catch (Exception ex) {

			logger.error(ex.getMessage());
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		} finally {
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	}


	/**
	 * 新增直播信息
	 *
	 * @param content
	 * @return
	 */
	private ConsumeConcurrentlyStatus anchorRelateLiveInsert(String tenantCode,String content) {
		try {
			if(tenantCode!=null&&StringUtils.isNotBlank(tenantCode))
			{
				TenantCodeUtil.getInstance().createServerTenantCode(tenantCode);
			}
			List<AnchorRelateLiveState> list = JSONArray.parseArray(content, AnchorRelateLiveState.class);
			if (list != null && list.size() > 0) {
				// 新增的集合
				List<AnchorRelateLiveState> insertList = new ArrayList<>();
				List<AnchorRelateLive> insertLiveList = new ArrayList<>();
				List<AnchorRelateLive> updateList = new ArrayList<>();
				// 根据状态和直播id查询有没有重复的 有则去重
				List<AnchorRelateLiveState> queryList = this.anchorRelateLiveStateService.selectLiveIds(tenantCode,list);

				//为了防止多节点时重复插入相同的记录数据
				StringBuilder keyStringBuilder=new StringBuilder();
				for (AnchorRelateLiveState anchorRelateLiveState : list) {
					keyStringBuilder.append(anchorRelateLiveState.getLiveId()+"-"+anchorRelateLiveState.getState());
				}
				String lockKey="anchorRelateLiveInsert_"+keyStringBuilder.toString();
				RedisUtil.getInstance().lock(lockKey,2000L);

				for (AnchorRelateLiveState anchorRelateLiveState : list) {
					// System.out.println("==anchorRelateLiveInsert=="+anchorRelateLiveState.getAnchorNickName()+"===="+anchorRelateLiveState.getAnchorId());
					boolean isInsert = true;
					for (AnchorRelateLiveState query : queryList) {
						if (anchorRelateLiveState.getLiveId().equals(query.getLiveId()) && anchorRelateLiveState.getState() == query.getState()) {
							isInsert = false;
							break;
						}
					}
					if (StringUtils.isNotBlank(anchorRelateLiveState.getAnchorId()) && isInsert) {
						anchorRelateLiveState.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
						anchorRelateLiveState.setCreateTime(new Date());
						insertList.add(anchorRelateLiveState);

						// 判断中间表是否要添加记录
						AnchorRelateLive queryAnchorRelateLive = this.anchorRelateLiveService.selectByLiveId(tenantCode,anchorRelateLiveState.getLiveId());
						if (queryAnchorRelateLive != null ) {
							if (queryAnchorRelateLive.getEndTime() == null && (anchorRelateLiveState.getState() == 2 || anchorRelateLiveState.getState() == 4)) {
								queryAnchorRelateLive.setEndTime(anchorRelateLiveState.getTime());
								queryAnchorRelateLive.setUpdateTime(new Date());
								if(queryAnchorRelateLive.getStartTime()!=null && queryAnchorRelateLive.getEndTime()!=null){
									// 计算时长
									BigDecimal liveTime = new BigDecimal((queryAnchorRelateLive.getEndTime().getTime()-queryAnchorRelateLive.getStartTime().getTime())/1000);
									// 如果小于0，则等于0
									if (liveTime.compareTo(BigDecimal.ZERO)<0) liveTime = BigDecimal.ZERO;
									queryAnchorRelateLive.setLiveTime(liveTime);
								}
								updateList.add(queryAnchorRelateLive);
							}
						} else {
							AnchorRelateLive anchorRelateLive = new AnchorRelateLive();
							anchorRelateLive.setLiveId(anchorRelateLiveState.getLiveId());
							if (anchorRelateLiveState.getState() == 1 || anchorRelateLiveState.getState() == 3) {
								anchorRelateLive.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
								anchorRelateLive.setAnchorId(anchorRelateLiveState.getAnchorId());
								anchorRelateLive.setAnchorNickName(anchorRelateLiveState.getAnchorNickName());
								anchorRelateLive.setFamilyId(anchorRelateLiveState.getFamilyId());
								anchorRelateLive.setRoomId(anchorRelateLiveState.getRoomId());
								anchorRelateLive.setStartTime(anchorRelateLiveState.getTime());
								anchorRelateLive.setCreateTime(new Date());
								anchorRelateLive.setType(anchorRelateLiveState.getType());
								for (AnchorRelateLiveState anchorRelateLiveState1 : list) {
									if (anchorRelateLiveState.getLiveId().equals(anchorRelateLiveState1.getLiveId()) && (anchorRelateLiveState1.getState() == 2 || anchorRelateLiveState1.getState() == 4)) {
										anchorRelateLive.setEndTime(anchorRelateLiveState1.getTime());
										if(anchorRelateLive.getStartTime()!=null && anchorRelateLive.getEndTime()!=null){
											// 计算时长
											BigDecimal liveTime = new BigDecimal((queryAnchorRelateLive.getEndTime().getTime()-queryAnchorRelateLive.getStartTime().getTime())/1000);
											// 如果小于0，则等于0
											if (liveTime.compareTo(BigDecimal.ZERO)<0) liveTime = BigDecimal.ZERO;
											anchorRelateLive.setLiveTime(liveTime);
										}
										break;
									}
								}
								insertLiveList.add(anchorRelateLive);
							}
						}
					}
				}

				// 一次性插入的数量
				int size = 500;
				// 总长度
				int length = insertList.size();
				// 计算可以分成多少组
				int num = (length + size - 1) / size;
				for (int i = 0; i < num; i++) {
					// 开始位置
					int fromIndex = i * size;
					// 结束位置
					int toIndex = (i + 1) * size < length ? (i + 1) * size : length;
					List<AnchorRelateLiveState> newAnchorRelateLiveStateList = insertList.subList(fromIndex, toIndex);
					this.anchorRelateLiveStateService.insertBatch(tenantCode,newAnchorRelateLiveStateList);
				}
				if (insertLiveList.size() > 0 || updateList.size() > 0) {
					this.anchorRelateLiveService.insertBatch(tenantCode,insertLiveList, updateList);
					this.anchorRelateLiveService.insertMonthBatch(tenantCode,insertLiveList, updateList);
				}
			}
		} catch (Exception ex) {

			logger.error(ex.getMessage());
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		} finally {
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	}



	/**
	 * 新增玩家消费信息
	 *
	 * @param content
	 * @return
	 */
	private ConsumeConcurrentlyStatus playerRelateConsumeInsert(String tenantCode,String content) {
		try {
			if(tenantCode!=null&&StringUtils.isNotBlank(tenantCode))
			{
				TenantCodeUtil.getInstance().createServerTenantCode(tenantCode);
			}
			List<PlayerRelateConsume> list = JSONArray.parseArray(content, PlayerRelateConsume.class);
			if (list != null && list.size() > 0) {
				// 新增的集合
				List<PlayerRelateConsume> insertList = new ArrayList<>();
				// 查询有没有重复的订单号 有则去重
				List<String> orderIds = list.stream().map(PlayerRelateConsume::getOrderId).collect(Collectors.toList());

				//为了防止多节点时重复插入相同的记录数据
				String lockKey0="playerRelateConsumeInsert_"+String.join(",",orderIds);
				RedisUtil.getInstance().lock(lockKey0,1000L);

				List<String> tempOrderIds =new ArrayList<>();
				if(orderIds!=null&&orderIds.size()>0) {
					tempOrderIds = this.playerRelateConsumeService.selectOrderIds(tenantCode, orderIds);
				}
				for (PlayerRelateConsume playerRelateConsume : list) {
					boolean isInsert = true;
					if(tempOrderIds!=null&&tempOrderIds.size()>0) {
						for (String orderId : tempOrderIds) {
							if (playerRelateConsume.getOrderId().equals(orderId)) {
								isInsert = false;
								break;
							}
						}
					}

					if (StringUtils.isNotBlank(playerRelateConsume.getPlayerId()) && isInsert) {
						playerRelateConsume.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
						playerRelateConsume.setCreateTime(new Date());
						// 主播信息,统计主播累充金额
						Anchor anchor = this.anchorService.selectByAnchorId(tenantCode,playerRelateConsume.getAnchorId());
						if(anchor!=null){
							playerRelateConsume.setRechargeIsUpdate(1);
							if(anchor.getAmount()!=null){
								anchor.setTotalAmount(anchor.getAmount().add(playerRelateConsume.getAmount()));
							}else {
								anchor.setTotalAmount(playerRelateConsume.getAmount());
							}
							this.anchorService.updateByPrimaryKeySelective(anchor);
						}
						//RedisUtil.getInstance().setForObject();
						// 玩家为主播累充信息表
						String playerId=playerRelateConsume.getPlayerId();
						String anchorId=playerRelateConsume.getAnchorId();
						if(StringUtils.isNotBlank(playerId)&&StringUtils.isNotBlank(anchorId)) {
							//防止在多线程情况下，同时执行插入方法
							//long random=(int)Math.random()*1000L;
							//Thread.sleep(1000+random);
							String lockKey="playerRelateAnchorRecharge_"+playerId+"_"+anchorId;
							try {
								//if (RedisUtil.getInstance().lock(lockKey,1000L))
								RedisUtil.getInstance().lock(lockKey,1000L);
								{
									PlayerRelateAnchorRecharge playerRelateAnchorRecharge = this.playerRelateAnchorRechargeService.selectByPlayerIdAndAnchorId(tenantCode, playerId, anchorId);
									if (playerRelateAnchorRecharge != null) {
										if (playerRelateConsume.getAmount() != null) {
											playerRelateAnchorRecharge.setAmount(playerRelateAnchorRecharge.getAmount().add(playerRelateConsume.getAmount()));
										}
										playerRelateAnchorRecharge.setUpdateTime(new Date());
										this.playerRelateAnchorRechargeService.updateByPrimaryKeySelective(tenantCode, playerRelateAnchorRecharge);
									} else {
										playerRelateAnchorRecharge = new PlayerRelateAnchorRecharge();
										playerRelateAnchorRecharge.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
										playerRelateAnchorRecharge.setAmount(playerRelateConsume.getAmount());
										playerRelateAnchorRecharge.setAnchorId(anchorId);
										playerRelateAnchorRecharge.setAnchorNickname(playerRelateConsume.getAnchorNickname());
										playerRelateAnchorRecharge.setChannelSubId(playerRelateConsume.getChannelSubId());
										playerRelateAnchorRecharge.setPlayerId(playerId);
										playerRelateAnchorRecharge.setPlayerNickname(playerRelateConsume.getPlayerNickname());
										InetAddress inetAddress = InetAddress.getLocalHost();
										playerRelateAnchorRecharge.setIp(inetAddress.getHostName()+"_"+inetAddress.getHostAddress());
										this.playerRelateAnchorRechargeService.insertSelective(tenantCode, playerRelateAnchorRecharge);
										//DistributedLockUtil.unLock(lockKey);
									}
								}
							}
							catch (Exception ex)
							{
								logger.error("=playerRelateAnchorRecharge="+ex+"=lockKey="+lockKey);
							}
						}
						insertList.add(playerRelateConsume);
					}
				}

				// 一次性插入的数量
				int size = 500;
				// 总长度
				int length = insertList.size();
				// 计算可以分成多少组
				int num = (length + size - 1) / size;
				for (int i = 0; i < num; i++) {
					// 开始位置
					int fromIndex = i * size;
					// 结束位置
					int toIndex = (i + 1) * size < length ? (i + 1) * size : length;
					List<PlayerRelateConsume> newPlayerRelateConsumeList = insertList.subList(fromIndex, toIndex);
					playerRelateConsumeService.insertBatch(tenantCode,newPlayerRelateConsumeList);
					playerRelateConsumeService.insertMonthBatch(tenantCode,newPlayerRelateConsumeList);
				}
			}
		} catch (Exception ex) {

			logger.error(ex.getMessage());
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		} finally {
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	}


	/**
	 * 新增玩家登录信息
	 *
	 * @param content
	 * @return
	 */
	private ConsumeConcurrentlyStatus playerRelateLoginInsert(String tenantCode,String content) {
		try {
			if(tenantCode!=null&&StringUtils.isNotBlank(tenantCode))
			{
				TenantCodeUtil.getInstance().createServerTenantCode(tenantCode);
			}
			List<PlayerRelateLogin> list = JSONArray.parseArray(content, PlayerRelateLogin.class);
			if (list != null && list.size() > 0) {
				// 新增集合
				List<PlayerRelateLogin> insertList = new ArrayList<>();
				// 新增3次登录集合
				List<ThreeLoginUser> insertThreeLoginList = new ArrayList<>();
				// 新增7次登录集合
				List<SevenLoginUser> insertSevenLoginList = new ArrayList<>();

				//为了防止多节点时重复插入相同的记录数据
				StringBuilder keyStringBuilder=new StringBuilder();
				for (PlayerRelateLogin playerRelateLogin : list) {
					keyStringBuilder.append(playerRelateLogin.getPlayerId()+"-"+playerRelateLogin.getLoginTime().getTime());
				}
				String lockKey0="playerRelateLoginInsert_"+keyStringBuilder.toString();
				RedisUtil.getInstance().lock(lockKey0,2000L);

				// 查询有没有重复记录
				List<PlayerRelateLogin> queryList = this.playerRelateLoginService.selectPlayerIds(tenantCode,list);
				for (PlayerRelateLogin playerRelateLogin : list) {
					boolean isInsert = true;
					for (PlayerRelateLogin query : queryList) {
						if (playerRelateLogin.getPlayerId().equals(query.getPlayerId()) && playerRelateLogin.getLoginTime().getTime() == query.getLoginTime().getTime()) {
							isInsert = false;
							break;
						}
					}
					if (StringUtils.isNotBlank(playerRelateLogin.getPlayerId()) && isInsert) {
						// 玩家信息，更新最后登录时间
						Player player = this.playerService.selectByPlayerId(tenantCode,playerRelateLogin.getPlayerId());
						if(player!=null){
							player.setLastLoginTime(playerRelateLogin.getLoginTime());
							this.playerService.updateByPrimaryKeySelective(tenantCode,player);
						}
						String lockKey="playerRelateLastLogin_"+playerRelateLogin.getPlayerId();
						try {
							//if (RedisUtil.getInstance().lock(lockKey,1000L))
							RedisUtil.getInstance().lock(lockKey,1000L);
							{
								// 玩家最后登录信息
								PlayerRelateLastLogin playerRelateLastLogin = this.playerRelateLastLoginService.selectByPlayerId(tenantCode, playerRelateLogin.getPlayerId());
								if (playerRelateLastLogin != null) {
									playerRelateLastLogin.setLoginCount(playerRelateLastLogin.getLoginCount() + 1);
									playerRelateLastLogin.setLoginTime(playerRelateLogin.getLoginTime());
									playerRelateLastLogin.setLoginOutTime(playerRelateLogin.getLoginOutTime());
									playerRelateLastLogin.setRegisterTime(playerRelateLogin.getRegisterTime());
									playerRelateLastLogin.setUpdateTime(new Date());
									this.playerRelateLastLoginService.updateByPrimaryKeySelective(tenantCode, playerRelateLastLogin);
								} else {
									PlayerRelateLastLogin insertPlayerRelateLastLogin = new PlayerRelateLastLogin();
									insertPlayerRelateLastLogin.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
									insertPlayerRelateLastLogin.setRegisterTime(playerRelateLogin.getRegisterTime());
									insertPlayerRelateLastLogin.setLoginOutTime(playerRelateLogin.getLoginOutTime());
									insertPlayerRelateLastLogin.setLoginTime(playerRelateLogin.getLoginTime());
									insertPlayerRelateLastLogin.setLoginCount(1);
									insertPlayerRelateLastLogin.setChannelId(playerRelateLogin.getChannelId());
									insertPlayerRelateLastLogin.setChannelSubId(playerRelateLogin.getChannelSubId());
									insertPlayerRelateLastLogin.setEquipment(playerRelateLogin.getEquipment());
									insertPlayerRelateLastLogin.setEquipmentCode(playerRelateLogin.getEquipmentCode());
									insertPlayerRelateLastLogin.setImei(playerRelateLogin.getImei());
									insertPlayerRelateLastLogin.setIp(playerRelateLogin.getIp());
									insertPlayerRelateLastLogin.setNickName(playerRelateLogin.getNickName());
									insertPlayerRelateLastLogin.setPlayerId(playerRelateLogin.getPlayerId());
									this.playerRelateLastLoginService.insertSelective(tenantCode, insertPlayerRelateLastLogin);
									//DistributedLockUtil.unLock(lockKey);
								}
							}
						}
						catch (Exception ex)
						{
                           logger.error("playerRelateLastLogin"+ex.getMessage());
						}

						// 玩家为主播累充信息表
						List<PlayerRelateAnchorRecharge> playerRelateAnchorRechargeList = this.playerRelateAnchorRechargeService.selectByPlayerId(tenantCode,playerRelateLogin.getPlayerId());
						if(playerRelateAnchorRechargeList!=null&&playerRelateAnchorRechargeList.size()>0){
							List<String> playerIdList=new ArrayList<>();
							for(int i=0;i<playerRelateAnchorRechargeList.size();i++) {
								playerIdList.add(playerRelateAnchorRechargeList.get(i).getPlayerId());
							}
							this.playerRelateAnchorRechargeService.batchUpdateByPlayerId(tenantCode,playerIdList,playerRelateLogin.getLoginTime(),new Date());
						}
						playerRelateLogin.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
						playerRelateLogin.setCreateTime(new Date());
						insertList.add(playerRelateLogin);

						Date loginTime=playerRelateLogin.getLoginTime();
						Date registerTime=playerRelateLogin.getLoginTime();
						int days=0;
						if(loginTime!=null&&registerTime!=null)
						{
							long between_days=(loginTime.getTime()-registerTime.getTime())/(1000*3600*24);
							days=Integer.parseInt(String.valueOf(between_days));
						}
						if(days==3)
						{
							insertThreeLoginList.add(getThreeLoginUser(playerRelateLogin));
						}
						else if(days==7)
						{
							insertSevenLoginList.add(getSevenLoginUser(playerRelateLogin));
						}
					}
				}

				insertLoginBatch(tenantCode,insertList);
				insertThreeBatch(tenantCode,insertThreeLoginList);
				insertSevenBatch(tenantCode,insertSevenLoginList);
			}
		} catch (Exception ex) {

			logger.error(ex.getMessage());
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		} finally {
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	}


	/**
	 * 批量新增登录用户信息
	 * @param tenantCode
	 * @param list
	 * @throws Exception
	 */
	private void insertLoginBatch(String tenantCode,List<PlayerRelateLogin> list) throws Exception
	{
		// 一次性插入的数量
		int size = 500;
		// 总长度
		int length = list.size();
		// 计算可以分成多少组
		int num = (length + size - 1) / size;
		for (int i = 0; i < num; i++) {
			// 开始位置
			int fromIndex = i * size;
			// 结束位置
			int toIndex = (i + 1) * size < length ? (i + 1) * size : length;
			List<PlayerRelateLogin> newPlayerRelateLoginList = list.subList(fromIndex, toIndex);
			playerRelateLoginService.insertBatch(tenantCode,newPlayerRelateLoginList);
		}
	}

	/**
	 * 批量新增三次登录用户信息
	 * @param tenantCode
	 * @param list
	 * @throws Exception
	 */
	private void insertThreeBatch(String tenantCode,List<ThreeLoginUser> list) throws Exception
	{
		// 一次性插入的数量
		int size = 500;
		// 总长度
		int length = list.size();
		// 计算可以分成多少组
		int num = (length + size - 1) / size;
		for (int i = 0; i < num; i++) {
			// 开始位置
			int fromIndex = i * size;
			// 结束位置
			int toIndex = (i + 1) * size < length ? (i + 1) * size : length;
			List<ThreeLoginUser> newThreeLoginList = list.subList(fromIndex, toIndex);
			threeLoginUserService.insertBatch(tenantCode,newThreeLoginList);
		}
	}


	/**
	 * 批量新增三次登录用户信息
	 * @param tenantCode
	 * @param list
	 * @throws Exception
	 */
	private void insertSevenBatch(String tenantCode,List<SevenLoginUser> list) throws Exception
	{
		// 一次性插入的数量
		int size = 500;
		// 总长度
		int length = list.size();
		// 计算可以分成多少组
		int num = (length + size - 1) / size;
		for (int i = 0; i < num; i++) {
			// 开始位置
			int fromIndex = i * size;
			// 结束位置
			int toIndex = (i + 1) * size < length ? (i + 1) * size : length;
			List<SevenLoginUser> newSevenLoginList = list.subList(fromIndex, toIndex);
			sevenLoginUserService.insertBatch(tenantCode,newSevenLoginList);
		}
	}


	/**
	 * 获取三次登录
	 * @param playerRelateLogin
	 * @return
	 */
	private ThreeLoginUser getThreeLoginUser(PlayerRelateLogin playerRelateLogin)
	{
		ThreeLoginUser threeLoginUser=new ThreeLoginUser();
		threeLoginUser.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
		threeLoginUser.setChannelSubId(playerRelateLogin.getChannelSubId());
		threeLoginUser.setPlayerId(playerRelateLogin.getPlayerId());
		threeLoginUser.setNickName(playerRelateLogin.getNickName());
		threeLoginUser.setAddTime(new Date());
		threeLoginUser.setCompletionTime(playerRelateLogin.getLoginTime());
		threeLoginUser.setCompletionTime(new Date());
		return threeLoginUser;
	}

	/**
	 * 获取三次登录
	 * @param playerRelateLogin
	 * @return
	 */
	private SevenLoginUser getSevenLoginUser(PlayerRelateLogin playerRelateLogin)
	{
		SevenLoginUser sevenLoginUser=new SevenLoginUser();
		sevenLoginUser.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
		sevenLoginUser.setChannelSubId(playerRelateLogin.getChannelSubId());
		sevenLoginUser.setPlayerId(playerRelateLogin.getPlayerId());
		sevenLoginUser.setNickName(playerRelateLogin.getNickName());
		sevenLoginUser.setAddTime(new Date());
		sevenLoginUser.setCompletionTime(playerRelateLogin.getLoginTime());
		sevenLoginUser.setCompletionTime(new Date());
		return sevenLoginUser;
	}

	/**
	 * 新增玩家充值信息
	 *
	 * @param content
	 * @return
	 */
	private ConsumeConcurrentlyStatus playerRelateRechargeInsert(String tenantCode,String content) {
		try {
			if(tenantCode!=null&&StringUtils.isNotBlank(tenantCode))
			{
				TenantCodeUtil.getInstance().createServerTenantCode(tenantCode);
			}
			List<PlayerRelateRecharge> list = JSONArray.parseArray(content, PlayerRelateRecharge.class);
			if (list != null && list.size() > 0) {
				// 新增集合
				List<PlayerRelateRecharge> insertList = new ArrayList<>();
				// 要修改累充金额的玩家
				List<Player> playerList = new ArrayList<>();
				// 查询有没有重复的订单号 有则去重
				List<String> orderIds = list.stream().map(PlayerRelateRecharge::getOrderId).collect(Collectors.toList());

				String lockKey0="playerRelateRechargeInsert_"+String.join(",",orderIds);
				RedisUtil.getInstance().lock(lockKey0,2000L);

				orderIds = this.playerRelateRechargeService.selectOrderIds(tenantCode,orderIds);

				// 记录累充金额
				Map<String,BigDecimal> map = new HashMap<>();
				for (PlayerRelateRecharge playerRelateRecharge : list) {
					boolean isInsert = true;
					for (String orderId : orderIds) {
						if (playerRelateRecharge.getOrderId().equals(orderId)) {
							isInsert = false;
							break;
						}
					}
					if (StringUtils.isNotBlank(playerRelateRecharge.getPlayerId()) && isInsert) {
						// 计算累充金额
						BigDecimal amount = map.get(playerRelateRecharge.getPlayerId());
						if(amount == null){
							amount = this.playerRelateRechargeService.selectAmountByPlayerId(tenantCode,playerRelateRecharge.getPlayerId());
						}
						if(amount!=null){
							amount = amount.add(playerRelateRecharge.getAmount());
							playerRelateRecharge.setTotalAmount(amount);
							map.put(playerRelateRecharge.getPlayerId(),amount);
						}else {
							playerRelateRecharge.setTotalAmount(playerRelateRecharge.getAmount());
							map.put(playerRelateRecharge.getPlayerId(),playerRelateRecharge.getAmount());
						}

						// 玩家信息,统计玩家累充金额
						Player player = this.playerService.selectByPlayerId(tenantCode,playerRelateRecharge.getPlayerId());
						if(player!=null){
							playerRelateRecharge.setRechargeIsUpdate(1);
							if(player.getAmount()!=null){
								player.setAmount(player.getAmount().add(playerRelateRecharge.getAmount()));
							}else {
								player.setAmount(playerRelateRecharge.getAmount());
							}
							this.playerService.updateByPrimaryKeySelective(tenantCode,player);
						}

						// 自带粉信息，统计自带粉累充金额
						AnchorRelateByoFans anchorRelateByoFans = this.anchorRelateByoFansService.selectAnchorRelateByoFansByPlayerId(tenantCode,playerRelateRecharge.getPlayerId());
						if(anchorRelateByoFans!=null){
							playerRelateRecharge.setBoysRechargeIsUpdate(1);
							if(anchorRelateByoFans.getAmount()!=null){
								anchorRelateByoFans.setAmount(anchorRelateByoFans.getAmount().add(playerRelateRecharge.getAmount()));
							}else {
								anchorRelateByoFans.setAmount(playerRelateRecharge.getAmount());
							}
							this.anchorRelateByoFansService.updateByPrimaryKeySelective(anchorRelateByoFans);
						}

						// 累充明细信息，统计业务员累充明细信息
						ParameterConfigure queryParam = this.parameterConfigureService.selectByCode(tenantCode,"CUMULATIVE_RECHARGE_TYPE");
						if (queryParam != null) {
							List<ParameterConfigure> parameterConfigureList = this.parameterConfigureService.selectListByParentId(tenantCode,queryParam.getId());
							if (parameterConfigureList.size() > 0) {
								// 子渠道总充金额
								BigDecimal total = this.playerService.selectAmountByChannelSubId(tenantCode,playerRelateRecharge.getChannelSubId());
								if(total!=null){
									for (ParameterConfigure parameterConfigure:parameterConfigureList) {
										String lockKey = "userCumulativeCharge_" + playerRelateRecharge.getChannelSubId() + "_" + parameterConfigure.getName();
										try {
											//if (RedisUtil.getInstance().lock(lockKey,1000L))
											RedisUtil.getInstance().lock(lockKey,1000L);
											{
												UserCumulativeCharge queryUserCum = this.userCumulativeChargeService.selectByChannelSubIdAndType(tenantCode, playerRelateRecharge.getChannelSubId(), parameterConfigure.getName());
												//if (queryUserCum == null && total.compareTo(new BigDecimal(parameterConfigure.getValue())) > 0)
												if (queryUserCum == null) {
													UserCumulativeCharge userCumulativeChargeAdd = new UserCumulativeCharge();
													userCumulativeChargeAdd.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
													userCumulativeChargeAdd.setType(parameterConfigure.getName());
													userCumulativeChargeAdd.setCompletionTime(playerRelateRecharge.getRechargeTime());
													userCumulativeChargeAdd.setChannelSubId(playerRelateRecharge.getChannelSubId());
													userCumulativeChargeAdd.setAmount(new BigDecimal(parameterConfigure.getValue()));
													userCumulativeChargeAdd.setCreateTime(new Date());
													this.userCumulativeChargeService.insertSelective(tenantCode, userCumulativeChargeAdd);
												}
											}
										}
										catch (Exception ex)
										{
											logger.error("=userCumulativeCharge="+ex.getMessage());
										}
									}
								}
							}
						}
						playerRelateRecharge.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
						playerRelateRecharge.setCreateTime(new Date());
						insertList.add(playerRelateRecharge);
					}
				}
				// 一次性插入的数量
				int size = 500;
				// 总长度
				int length = insertList.size();
				// 计算可以分成多少组
				int num = (length + size - 1) / size;
				for (int i = 0; i < num; i++) {
					// 开始位置
					int fromIndex = i * size;
					// 结束位置
					int toIndex = (i + 1) * size < length ? (i + 1) * size : length;
					List<PlayerRelateRecharge> newPlayerRelateRechargeList = insertList.subList(fromIndex, toIndex);
					playerRelateRechargeService.insertBatch(tenantCode,newPlayerRelateRechargeList);
					playerRelateRechargeService.insertMonthBatch(tenantCode,newPlayerRelateRechargeList);
				}
			}
		} catch (Exception ex) {

			logger.error(ex.getMessage());
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		} finally {
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	}



	/**
	 * 新增玩家观看直播信息
	 *
	 * @param content
	 * @return
	 */
	private ConsumeConcurrentlyStatus playerRelateViewLiveInsert(String tenantCode,String content) {
		try {
			if(tenantCode!=null&&StringUtils.isNotBlank(tenantCode))
			{
				TenantCodeUtil.getInstance().createServerTenantCode(tenantCode);
			}
			List<PlayerRelateViewLiveState> list = JSONArray.parseArray(content, PlayerRelateViewLiveState.class);
			if (list != null && list.size() > 0) {
				// 新增集合
				List<PlayerRelateViewLiveState> insertList = new ArrayList<>();
				List<PlayerRelateViewLive> liveInsertList = new ArrayList<>();
				List<PlayerRelateViewLive> updateList = new ArrayList<>();

				//为了防止多节点时重复插入相同的记录数据
				StringBuilder keyStringBuilder=new StringBuilder();
				for (PlayerRelateViewLiveState playerRelateViewLiveState : list) {
					keyStringBuilder.append(playerRelateViewLiveState.getLiveId()+"-"+playerRelateViewLiveState.getState());
				}
				String lockKey0="playerRelateViewLiveInsert_"+keyStringBuilder.toString();
				RedisUtil.getInstance().lock(lockKey0,2000L);

				// 查询有没有重复的 有则去重
				List<PlayerRelateViewLiveState> queryList = this.playerRelateViewLiveStateService.selectLiveIds(tenantCode,list);
				for (PlayerRelateViewLiveState playerRelateViewLiveState : list) {
					boolean isInsert = true;
					for (PlayerRelateViewLiveState query : queryList) {
						if (playerRelateViewLiveState.getLiveId().equals(query.getLiveId()) && playerRelateViewLiveState.getState() == query.getState()) {
							isInsert = false;
							break;
						}
					}
					if (StringUtils.isNotBlank(playerRelateViewLiveState.getPlayerId()) && isInsert) {
						playerRelateViewLiveState.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
						playerRelateViewLiveState.setCreateTime(new Date());
						insertList.add(playerRelateViewLiveState);

						// 判断中间表是否要添加记录
						PlayerRelateViewLive queryPlayerRelateViewLive = this.playerRelateViewLiveService.selectByLiveId(tenantCode,playerRelateViewLiveState.getLiveId());
						if (queryPlayerRelateViewLive != null) {
							if (queryPlayerRelateViewLive.getEndTime() == null && (playerRelateViewLiveState.getState() == 2 || playerRelateViewLiveState.getState() == 4)) {
								queryPlayerRelateViewLive.setEndTime(playerRelateViewLiveState.getTime());
								queryPlayerRelateViewLive.setUpdateTime(new Date());
								// 计算观看直播时长
								if(queryPlayerRelateViewLive.getStartTime()!=null && queryPlayerRelateViewLive.getEndTime()!=null){
									BigDecimal liveTime = new BigDecimal((queryPlayerRelateViewLive.getEndTime().getTime()-queryPlayerRelateViewLive.getStartTime().getTime())/1000);
									// 如果小于0，则等于0
									if (liveTime.compareTo(BigDecimal.ZERO)<0) liveTime = BigDecimal.ZERO;
									queryPlayerRelateViewLive.setLiveTime(liveTime);
									Player player = this.playerService.selectByPlayerId(tenantCode,queryPlayerRelateViewLive.getPlayerId());
									if(player!=null){
										if(player.getLiveTime()!=null){
											player.setLiveTime(player.getLiveTime().add(liveTime));
										}else {
											player.setLiveTime(liveTime);
										}
										this.playerService.updateByPrimaryKeySelective(tenantCode,player);
									}
								}
								updateList.add(queryPlayerRelateViewLive);
							}
						} else {
							PlayerRelateViewLive playerRelateViewLive = new PlayerRelateViewLive();
							playerRelateViewLive.setLiveId(playerRelateViewLiveState.getLiveId());
							if (playerRelateViewLiveState.getState() == 1 || playerRelateViewLiveState.getState() == 3) {
								playerRelateViewLive.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
								playerRelateViewLive.setAnchorId(playerRelateViewLiveState.getAnchorId());
								playerRelateViewLive.setPlayerId(playerRelateViewLiveState.getPlayerId());
								playerRelateViewLive.setChannelId(playerRelateViewLiveState.getChannelId());
								playerRelateViewLive.setChannelSubId(playerRelateViewLiveState.getChannelSubId());
								playerRelateViewLive.setRoomId(playerRelateViewLiveState.getRoomId());
								playerRelateViewLive.setStartTime(playerRelateViewLiveState.getTime());
								playerRelateViewLive.setCreateTime(new Date());
								for (PlayerRelateViewLiveState playerRelateViewLiveState1 : list) {
									if (playerRelateViewLiveState.getLiveId().equals(playerRelateViewLiveState1.getLiveId()) && (playerRelateViewLiveState1.getState() == 2 || playerRelateViewLiveState1.getState() == 4)) {
										playerRelateViewLive.setEndTime(playerRelateViewLiveState1.getTime());
										// 计算观看直播时长
										if(playerRelateViewLive.getStartTime()!=null && playerRelateViewLive.getEndTime()!=null){
											BigDecimal liveTime = new BigDecimal((playerRelateViewLive.getEndTime().getTime()-playerRelateViewLive.getStartTime().getTime())/1000);
											// 如果小于0，则等于0
											if (liveTime.compareTo(BigDecimal.ZERO)<0) liveTime = BigDecimal.ZERO;
											playerRelateViewLive.setLiveTime(liveTime);
											Player player = this.playerService.selectByPlayerId(tenantCode,playerRelateViewLive.getPlayerId());
											if(player!=null){
												if(player.getLiveTime()!=null){
													player.setLiveTime(player.getLiveTime().add(liveTime));
												}else {
													player.setLiveTime(liveTime);
												}
												this.playerService.updateByPrimaryKeySelective(tenantCode,player);
											}
										}
										break;
									}
								}
								liveInsertList.add(playerRelateViewLive);
							}
						}
					}
				}
				// 一次性插入的数量
				int size = 500;
				// 总长度
				int length = insertList.size();
				// 计算可以分成多少组
				int num = (length + size - 1) / size;
				for (int i = 0; i < num; i++) {
					// 开始位置
					int fromIndex = i * size;
					// 结束位置
					int toIndex = (i + 1) * size < length ? (i + 1) * size : length;
					List<PlayerRelateViewLiveState> newPlayerRelateViewLivestateList = insertList.subList(fromIndex, toIndex);
					playerRelateViewLiveStateService.insertBatch(tenantCode,newPlayerRelateViewLivestateList);
				}

				if (liveInsertList.size() > 0 || updateList.size() > 0) {
					this.playerRelateViewLiveService.insertBatch(tenantCode,liveInsertList, updateList);
					this.playerRelateViewLiveService.insertMonthBatch(tenantCode,liveInsertList, updateList);
				}
			}
		} catch (Exception ex) {

			logger.error(ex.getMessage());
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		} finally {
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	}


	/**
	 * 新增打招呼信息
	 *
	 * @param content
	 * @return
	 */
	private ConsumeConcurrentlyStatus anchorRelateInteractInsert(String tenantCode,String content) {
		try {
			if(tenantCode!=null&&StringUtils.isNotBlank(tenantCode))
			{
				TenantCodeUtil.getInstance().createServerTenantCode(tenantCode);
			}
			List<AnchorRelateInteract> list = JSONArray.parseArray(content, AnchorRelateInteract.class);
			if (list != null && list.size() > 0) {
				// 新增集合
				List<AnchorRelateInteract> insertList = new ArrayList<>();
				//为了防止多节点时重复插入相同的记录数据
				StringBuilder keyStringBuilder=new StringBuilder();
				for (AnchorRelateInteract anchorRelateInteract : list) {
					keyStringBuilder.append(anchorRelateInteract.getFromId()+"-"+anchorRelateInteract.getTime().getTime());
				}
				String lockKey0="anchorRelateInteractInsert_"+keyStringBuilder.toString();
				RedisUtil.getInstance().lock(lockKey0,1000L);

				// 查询有没有重复的记录 有则去重
				List<AnchorRelateInteract> queryList = this.anchorRelateInteractService.selectFormIds(tenantCode,list);
				for (AnchorRelateInteract anchorRelateInteract : list) {
					boolean isInsert = true;
					for (AnchorRelateInteract query : queryList) {
						if (anchorRelateInteract.getFromId().equals(query.getFromId()) && anchorRelateInteract.getTime().getTime() == query.getTime().getTime()) {
							isInsert = false;
							break;
						}
					}
					if (StringUtils.isNotBlank(anchorRelateInteract.getFromId()) && isInsert) {
						anchorRelateInteract.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
						anchorRelateInteract.setCreateTime(new Date());
						insertList.add(anchorRelateInteract);
					}
				}

				// 一次性插入的数量
				int size = 500;
				// 总长度
				int length = insertList.size();
				// 计算可以分成多少组
				int num = (length + size - 1) / size;
				for (int i = 0; i < num; i++) {
					// 开始位置
					int fromIndex = i * size;
					// 结束位置
					int toIndex = (i + 1) * size < length ? (i + 1) * size : length;
					List<AnchorRelateInteract> newAnchorRelateInteractList = insertList.subList(fromIndex, toIndex);
					anchorRelateInteractService.insertBatch(tenantCode,newAnchorRelateInteractList);
				}
			}
		} catch (Exception ex) {

			logger.error(ex.getMessage());
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		} finally {
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	}


	/**
	 * 新增自带粉信息
	 *
	 * @param content
	 * @return
	 */
	private ConsumeConcurrentlyStatus anchorRelateByoFansInsert(String tenantCode,String content) {
		try {
			if(tenantCode!=null&&StringUtils.isNotBlank(tenantCode))
			{
				TenantCodeUtil.getInstance().createServerTenantCode(tenantCode);
			}
			List<AnchorRelateByoFans> list = JSONArray.parseArray(content, AnchorRelateByoFans.class);
			if (list != null && list.size() > 0) {
				// 新增集合
				List<AnchorRelateByoFans> insertList = new ArrayList<>();

				//为了防止多节点时重复插入相同的记录数据
				StringBuilder keyStringBuilder=new StringBuilder();
				for (AnchorRelateByoFans anchorRelateByoFans : list) {
					keyStringBuilder.append(anchorRelateByoFans.getAnchorId()+"-"+anchorRelateByoFans.getByoFansId());
				}
				String lockKey0="anchorRelateByoFansInsert_"+keyStringBuilder.toString();
				RedisUtil.getInstance().lock(lockKey0,1000L);

				// 查询有没有重复的记录 有则去重
				List<AnchorRelateByoFans> queryList = this.anchorRelateByoFansService.selectByoFansId(tenantCode,list);
				for (AnchorRelateByoFans anchorRelateByoFans : list) {
					boolean isInsert = true;
					for (AnchorRelateByoFans query : queryList) {
						if (anchorRelateByoFans.getAnchorId().equals(query.getAnchorId()) && anchorRelateByoFans.getByoFansId() == query.getByoFansId()) {
							isInsert = false;
							break;
						}
					}
					if (StringUtils.isNotBlank(anchorRelateByoFans.getAnchorId()) && isInsert) {
						anchorRelateByoFans.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
						anchorRelateByoFans.setCreateTime(new Date());
						insertList.add(anchorRelateByoFans);
					}
				}

				// 一次性插入的数量
				int size = 500;
				// 总长度
				int length = insertList.size();
				// 计算可以分成多少组
				int num = (length + size - 1) / size;
				for (int i = 0; i < num; i++) {
					// 开始位置
					int fromIndex = i * size;
					// 结束位置
					int toIndex = (i + 1) * size < length ? (i + 1) * size : length;
					List<AnchorRelateByoFans> newAnchorRelateByoFansList = insertList.subList(fromIndex, toIndex);
					anchorRelateByoFansService.insertBatch(newAnchorRelateByoFansList);
				}
			}
		} catch (Exception ex) {

			logger.error(ex.getMessage());
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		} finally {
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	}


	/**
	 * 新增玩家渠道改绑记录
	 *
	 * @param content
	 * @return
	 */
	private ConsumeConcurrentlyStatus playerRelateChannelLogInsert(String tenantCode,String content) {
		try {
			if(tenantCode!=null&&StringUtils.isNotBlank(tenantCode))
			{
				TenantCodeUtil.getInstance().createServerTenantCode(tenantCode);
			}
			List<PlayerRelateChannelLog> list = JSONArray.parseArray(content, PlayerRelateChannelLog.class);
			if (list != null && list.size() > 0) {
				// 新增集合
				List<PlayerRelateChannelLog> insertList = new ArrayList<>();

				//为了防止多节点时重复插入相同的记录数据
				StringBuilder keyStringBuilder=new StringBuilder();
				for (PlayerRelateChannelLog playerRelateChannelLog : list) {
					keyStringBuilder.append(playerRelateChannelLog.getChannelSubId()+"-"+playerRelateChannelLog.getOldChannelSubId()+"-"+playerRelateChannelLog.getChangeChannelTime().getTime());
				}
				String lockKey0="playerRelateChannelLogInsert_"+keyStringBuilder.toString();
				RedisUtil.getInstance().lock(lockKey0,1000L);

				// 查询有没有重复的记录 有则去重
				List<PlayerRelateChannelLog> queryList = this.playerRelateChannelLogService.selectChannelSubIdAndOldChannelSubId(tenantCode,list);
				for (PlayerRelateChannelLog playerRelateChannelLog : list) {
					// 青芒那边传过来的字段不一致，所以要赋值
					playerRelateChannelLog.setChannelSubId(playerRelateChannelLog.getSubChannelId());
					playerRelateChannelLog.setOldChannelSubId(playerRelateChannelLog.getOldSubChannelId());
					boolean isInsert = true;
					for (PlayerRelateChannelLog query : queryList) {
						if (playerRelateChannelLog.getChannelSubId().equals(query.getChannelSubId()) && playerRelateChannelLog.getOldChannelSubId().equals(query.getOldChannelSubId()) && playerRelateChannelLog.getChangeChannelTime().getTime() == query.getChangeChannelTime().getTime()) {
							isInsert = false;
							break;
						}
					}
					if (StringUtils.isNotBlank(playerRelateChannelLog.getPlayerId()) && isInsert) {
						playerRelateChannelLog.setId(PrimaryKeyGeneratorUtil.getInstance().getRandomKey());
						playerRelateChannelLog.setCreateTime(new Date());
						playerRelateChannelLog.setSource(1);
						playerRelateChannelLog.setIsUpdate(0);
						insertList.add(playerRelateChannelLog);
					}
				}

				// 一次性插入的数量
				int size = 500;
				// 总长度
				int length = insertList.size();
				// 计算可以分成多少组
				int num = (length + size - 1) / size;
				for (int i = 0; i < num; i++) {
					// 开始位置
					int fromIndex = i * size;
					// 结束位置
					int toIndex = (i + 1) * size < length ? (i + 1) * size : length;
					List<PlayerRelateChannelLog> newPlayerRelateChannelLog = insertList.subList(fromIndex, toIndex);
					playerRelateChannelLogService.insertBatch(tenantCode,newPlayerRelateChannelLog);
				}
			}
		} catch (Exception ex) {

			logger.error(ex.getMessage());
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		} finally {
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	}
}
